%%% -------------------------------------------------------------------
%%% Author  : vjache
%%% Description : Beta node activation server. It is responsible for 
%%%				  modification of beta memory of particular node.
%%%
%%% Created : Jan 10, 2010
%%% -------------------------------------------------------------------
-module(beta_node_act_srv).

-behaviour(gen_server).
%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------
-include_lib("eunit/include/eunit.hrl").
-include("../../include/erete.hrl").
-include("../../include/pipe.hrl").
-include("../../include/iter.hrl").
%% --------------------------------------------------------------------
%% External exports
%% --------------------------------------------------------------------
-export([start_link/1,server_name/1]).

%% --------------------------------------------------------------------
%% gen_server callbacks
%% --------------------------------------------------------------------
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

%%--------------------------------------------------------------------
%% Record representing a state of beta node activation server.
%%--------------------------------------------------------------------
-record(state, 
		{
		 filename,	% File name to be used when reopen log
		 beta_mem,	% A beta memory handle
		 conf		% Beta node configuration
		}).

%% ====================================================================
%% External functions
%% ====================================================================
start_link(Conf) when ?IS_BETA_NODE_CONF_VALID(Conf)->
	SrvName=server_name(Conf#beta_node_conf.id),
    gen_server:start_link(
	  {local, SrvName }, 
	  ?MODULE,
	  [Conf], 
	  []);
start_link(BadConf) ->
	{error,{bad_conf,BadConf}}.

server_name(BetaNodeId) when is_atom(BetaNodeId) ->
	list_to_atom(atom_to_list(BetaNodeId)++"act").

%% ====================================================================
%% Server functions
%% ====================================================================

%% --------------------------------------------------------------------
%% Function: init/1
%% Description: Initiates the server
%% Returns: {ok, State}          |
%%          {ok, State, Timeout} |
%%          ignore               |
%%          {stop, Reason}
%% --------------------------------------------------------------------
init([Conf]) ->
	% 1. Get conventional log name (currently beta node id)
	LogName=Conf#beta_node_conf.id,
	% 2. Form name of file rename log to
	File="./"++atom_to_list(LogName)++"_r.LOG",
	% 3. Call beta memory factory fun in server process
	%	 giving a chance to link underlying resources
	BetaMemFun=Conf#beta_node_conf.beta_mem_fun,
	Bmh=BetaMemFun(Conf#beta_node_conf.id),
    {ok, #state{filename=File,beta_mem=Bmh,conf=Conf},5000}.

%% --------------------------------------------------------------------
%% Function: handle_call/3
%% Description: Handling call messages
%% Returns: {reply, Reply, State}          |
%%          {reply, Reply, State, Timeout} |
%%          {noreply, State}               |
%%          {noreply, State, Timeout}      |
%%          {stop, Reason, Reply, State}   | (terminate/2 is called)
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_call(_Request, _From, State) ->
    Reply = not_supported,
    {reply, Reply, State}.

%% --------------------------------------------------------------------
%% Function: handle_cast/2
%% Description: Handling cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_cast(Msg, State) ->
	try cast(Msg, State) of
		Ret -> Ret
	catch
		throw:Exc ->
			error_logger:error_report([
				  {message,Msg},
				  {state,State},
				  {reason,Exc},
				  {stacktrace,erlang:get_stacktrace()}]),
			{stop, Exc, State}
	end.

%% --------------------------------------------------------------------
%% Function: handle_info/2
%% Description: Handling all non call/cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_info(timeout, State) ->
	{noreply, State, hibernate};
handle_info(_Info, State) ->
	{noreply, State}.

%% --------------------------------------------------------------------
%% Function: terminate/2
%% Description: Shutdown the server
%% Returns: any (ignored by gen_server)
%% --------------------------------------------------------------------
terminate(_Reason, _State) ->
    ok.

%% --------------------------------------------------------------------
%% Func: code_change/3
%% Purpose: Convert process state when code is changed
%% Returns: {ok, NewState}
%% --------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% --------------------------------------------------------------------
%%% Internal functions
%% --------------------------------------------------------------------

beta_memory_modify(#mdf_req{mdf_code=ins,input_id=MpId,data={K,V}},BetaMemHandle) ->
	beta_memory:insert(BetaMemHandle,MpId,K,V);
beta_memory_modify(#mdf_req{mdf_code=del,input_id=MpId,data={K,V}},BetaMemHandle) ->
	beta_memory:delete(BetaMemHandle,MpId,K,V).
%% --------------------------------------------------------------------
%% Func: activation_init(LogIter,State) -> ok
%% Purpose: Entry point to the activation routine.
%% Returns: ok
%% --------------------------------------------------------------------
activation_init(LogIter,State) when 
  ?IS_ITER(LogIter),is_record(State,state) ->
	activation_loop(iter:open(LogIter),State).
%% --------------------------------------------------------------------
%% Func: activation_loop(LogIterState,State) -> ok
%% Purpose: Traverses beta memory modification requests, 
%%			produces solutions, reduces them and feeds 
%% 			them to the output pipe.
%% Returns: ok
%% --------------------------------------------------------------------
activation_loop(LogIterState,State) ->
	Conf=State#state.conf,
	Inputs=Conf#beta_node_conf.in_ids,
	OutPipe=Conf#beta_node_conf.out_pipe,
	BetaMem=State#state.beta_mem,
	RedFun=Conf#beta_node_conf.red_fun,
	RedMode=Conf#beta_node_conf.red_mode,
	% 1. Take item from log
	{LogIterState1,Msg}=iter:next(LogIterState),
	% 2. Determine modification request
	case Msg of
		% 2.a. End of log reached
		?EOF -> iter:close(LogIterState1);
		% 2.b. Insert|delete request handling
		#mdf_req{mdf_code=Modif,input_id=MpId,data={K,V}} ->
			% 2.b.1. Modify beta memory
			case beta_memory_modify(Msg,BetaMem) of
				% 2.b.1.a Memory IS modified
				true ->
					% 2.b.1.a.1 Form list with parts of cartesian product
					Iters=[ case MpId1 of
								MpId when RedMode==inc ->
									iter:single(V);
								_ ->
									beta_memory:fetch_values(BetaMem,MpId1,K)
							end ||MpId1<-Inputs],
					% 2.b.1.a.2 Reduce cartesian product
					ReducedIter=RedFun(Modif,K,Iters),
					% 2.b.1.a.3 Push reduced solutions to the output.
					%			Solutions marked with modification tag (ins|del).
					Pump=pump:new(iter_mapper:new(ReducedIter, 
											 fun(Item)->
													 {Modif,Item}
											 end), 
							 OutPipe),
					pump:init(Pump),
					% 2.b.1.a.4 Do next request
					activation_loop(LogIterState1,State);
				% 2.b.1.b Memory IS NOT modified
				false -> 
					% 2.b.1.b.1 Do next request
					activation_loop(LogIterState1,State)
			end;
		% 2.c. Unsupported modification request handling
		BadModif ->
			throw({unsup_modif,BadModif})
	end.

cast({activate_from_log,OnDoneCallback},State) ->
	% activation logic here
	Conf=State#state.conf,
	% 1. Get conventional log name (currently beta node id)
	LogName=Conf#beta_node_conf.id,
	cast({activate_from_log,OnDoneCallback,{name,LogName}},State);
cast({activate_from_log,OnDoneCallback,{name,LogName}},State) ->
	% activation logic here
	% 1. Open log for iteration
	Iter=iter_disk_log:new(
		   [{name,LogName}]),
	% 2. Do activation
	cast({activate_from_iter,OnDoneCallback,Iter},State);
cast({activate_from_log,OnDoneCallback,{filename,LogFileName}},State) ->
	% activation logic here
	% 1. Open log for iteration
	Iter=iter_disk_log:new(
		   [{name,LogFileName},{file,LogFileName}]),
	% 2. Do activation
	cast({activate_from_iter,OnDoneCallback,Iter},State);
cast({activate_from_iter,OnDoneCallback,SrcIter},State) ->
	% activation logic here
	Iter=mdf_req_iter(SrcIter,State),
	try activation_init(Iter, State) of
		_ ->
			case OnDoneCallback of
				noop -> ok;
				_ -> OnDoneCallback(ok)
			end
	catch
		throw:Reason ->
			case OnDoneCallback of
				noop -> ok;
				_ -> OnDoneCallback({error,Reason})
			end,
			throw({activation_failed,Reason,erlang:get_stacktrace()})
	end,
	{noreply, State, 5000};
cast(Msg, State) ->
	throw({bad_cast,Msg,State}).

%%----------------------------------------------------------
%% Function: mdf_req_iter(BaseIter,State) -> iter()
%% Purpose: Wraps base iter BaseIter with iter that:
%% 			1. Flattens batch mdf requests i.e. 
%% 				#mdf_req{data=[{K,V}|_]} -> [#mdf_req{data={K,V}}]
%% 			2. Validates mdf requests.
%% Returns: iterator over #mdf_req's
%%---------------------------------------------------------- 
mdf_req_iter(BaseIter,State) when 
  ?IS_ITER(BaseIter),is_record(State,state) ->
	Conf=State#state.conf,
	InpIds=Conf#beta_node_conf.in_ids,
	% Check function
	InpCheck=fun(Item)->
					 InpId=Item#mdf_req.input_id,
					 case lists:member(InpId, InpIds) of
						 true -> ok;
						 false ->
							 throw({bad_input_id,Item,InpIds})
					 end
			 end,
	iter_concat:new(
	  iter_mapper:new(
		BaseIter, 
		fun
			%% Check & flatten #mdf_req 		   
		   (Item=#mdf_req{input_id=InpId,mdf_code=Code,data=KVList})
			 when is_list(KVList)->
				InpCheck(Item),
				iter_mapper:new(iter:from_list(KVList), 
								fun({K,V})->
										#mdf_req{input_id=InpId,mdf_code=Code,data={K,V}}
								end);
		   %% Just check
		   (Item=#mdf_req{input_id=_InpId,mdf_code=_Code,data={_K,_V}})->
				InpCheck(Item),
				Item;
		   %% Raise exception
		   (BadReq)->
				throw({bad_mdf_req,BadReq})
		end)).

%% --------------------------------------------------------------------
%%% eUnit test functions
%% --------------------------------------------------------------------
activate_from_iter_test() ->
	Conf=#beta_node_conf
		 {
		  id='bn#1', 									% an atom identifying beta_node server
		  in_ids=['an#1','an#2'],		 				% a list of atoms
		  out_pipe=pipe_console:new(), 					% an output pipe
		  beta_mem_fun=fun beta_memory_ets:open/1,		% beta memory factory fun -> #mh
		  red_fun=fun(_Modif,_K,IterList)->				% reduce function 
						  iter_join_cprod:all(IterList)
				  end,
		  red_mode = inc								% reduce mode (inc|aggr)
		 },
	{ok,_}=start_link(Conf),
	ServerRef=server_name('bn#1'),
	Mdf1=#mdf_req{input_id='an#1',
			 mdf_code=ins,
			 data=[{k1,v1},{k2,v2},{k3,v3}]},
	gen_server:cast(ServerRef, {activate_from_iter,noop,iter:single(Mdf1)}),
	Mdf2=#mdf_req{input_id='an#2',
			 mdf_code=ins,
			 data=[{k1,v11},{k2,v22}]},
	gen_server:cast(ServerRef, {activate_from_iter,noop,iter:single(Mdf2)}),
	Mdf3=#mdf_req{input_id='an#2',
			 mdf_code=ins,
			 data=[{k1,v21},{k3,v33}]},
	gen_server:cast(ServerRef, {activate_from_iter,noop,iter:single(Mdf3)}).

activate_from_log_test() ->
	Conf=#beta_node_conf
		 {
		  id=Id='bn#1', 									% an atom identifying beta_node server
		  in_ids=['an#1','an#2'],		 				% a list of atoms
		  out_pipe=pipe_console:new(), 					% an output pipe
		  beta_mem_fun=fun beta_memory_ets:open/1,		% beta memory factory fun -> #mh
		  red_fun=fun(_Modif,_K,IterList)->				% reduce function 
						  iter_join_cprod:all(IterList)
				  end,
		  red_mode = inc								% reduce mode (inc|aggr)
		 },
	Pipe=pipe_disk_logger:new([{name,Id}]),
	Iter=iter:from_list([
	#mdf_req{input_id='an#1',
			 mdf_code=ins,
			 data=[{k1,v1},{k2,v2},{k3,v3}]},
	#mdf_req{input_id='an#3',
			 mdf_code=ins,
			 data=[{k1,v11},{k2,v22}]}]),
	Pump=pump:new(Iter, Pipe),
	pump:init(Pump),
	{ok,_}=start_link(Conf),
	ServerRef=server_name(Id),
	OnDoneCallback=fun(ok)->
						   file:delete("./"++atom_to_list(Id)++".LOG");
					  ({error,Reason}) ->
						   io:format("Activation error ~p~n",[Reason])
				   end,
	gen_server:cast(ServerRef, {activate_from_log,OnDoneCallback,{name,Id}}),
	ok.
